#include "threadpool.h"

unsigned int CoreCount()
{
    unsigned count = 1; // 至少一个
    #if defined (LINUX)
    count = sysconf(_SC_NPROCESSORS_CONF);
    #elif defined (WINDOWS)
    SYSTEM_INFO si;
    GetSystemInfo(&si);
    count = si.dwNumberOfProcessors;
    #endif
    return count;
}

namespace general{
    CThreadPool::CThreadPool(int num) {
        this->mThreadCnt = num;
        mStarted = false;
    }

    int CThreadPool::AddTask(Task *t) {
        if ( nullptr == t){
            return -1;
        }
        if(!mStoping){
            {
                std::unique_lock<std::mutex> lck(this->mMutex);
                this->mTasks.push(t);
            }
            mQueueAvaliableCondition.notify_one();
            return 0;
        }else{
            return -2;
        }
    }

    void CThreadPool::Start() {
        mStarted = true;
        mThreads.reserve(this->mThreadCnt);
        for (int i = 0; i < mThreadCnt; i++)
        {
            mThreads.emplace_back(new std::thread(&ThreadPool::Process, this, i));
        }
    }

    Task *CThreadPool::PopTask() {
        std::unique_lock<std::mutex> lk(this->mMutex);
        while (mTasks.empty() && mStarted)
        {
            mQueueAvaliableCondition.wait(lk);
        }
        if(!mStarted){
            return nullptr;
        }
        Task *ret = this->mTasks.front();
        this->mTasks.pop();
        return ret;
    }

    void CThreadPool::Process(int id) {
        while (mStarted)
        {
            Task *task = PopTask();
            if (nullptr == task)
            {
                continue;
            }
            else
            {
                task->Run();
                task->OnDone();
            }
        }
    }

    void CThreadPool::Stop() {
        {
            std::lock_guard<std::mutex> lk(mMutex);
            mStarted = false;
            mQueueAvaliableCondition.notify_all();
        }

        for (auto &th : mThreads)
        {
            th->join();
        }
        mStarted = false;
    }
    void CThreadPool::StopAll() {
        {
            while(this->mTasks.size() > 0) 
                std::this_thread::sleep_for(std::chrono::milliseconds(100));
            mStarted = false;
            mQueueAvaliableCondition.notify_all();
        }

        for (auto &th : mThreads)
        {
            th->join();
        }
        mStarted = false;
    }
    bool CThreadPool::Started(){
        return this->mStarted;
    }
    CThreadPool::~CThreadPool(){
        if (this->mStarted)
        {
            std::cout << "desruction" << std::endl;
            this->mStarted = false;
            {
                std::lock_guard<std::mutex> lk(mMutex);
                mStarted = false;
                mQueueAvaliableCondition.notify_all();
            }
            for (size_t i = 0; i != this->mThreads.size(); ++i)
            {
                if (mThreads[i]->joinable())
                {
                    mThreads[i]->join();
                }
            }
        }
    }
}

